package com.zx._03_并发.chapter_06_compent;

import com.google.common.collect.Lists;
import com.zx.common.util.ThreadUtil;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import lombok.Builder;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

// 案例大全：
//			https://mp.weixin.qq.com/s/VGDlhLRSsQ7X63R1fKmLdQ
// 使用大全: https://blog.csdn.net/qq_31865983/article/details/106137777
@Slf4j
public class Demo_10_CompletableFutureDemo {

    public static ExecutorService executor = Executors.newFixedThreadPool(3, new ThreadFactory() {
        private AtomicInteger count = new AtomicInteger();

        @Override
        public Thread newThread(final Runnable runnable) {
            return new Thread(runnable, "custom-executor-" + count.incrementAndGet());
        }
    });

    /**
     * 创建一个已经完成的CompletableFuture（其实就是个同步的数据），没有意义的代码
     */
    @Test
    public void test01() {

        final CompletableFuture<String> future = CompletableFuture.completedFuture("消息");

        final boolean done = future.isDone();
        log.info("{}", done);

        final String now = future.getNow("没有数据");//立刻获得数据，没有获得数据的话，就返回valueIfAbsent
        log.info(now);
    }

    /**
     * 创建一个异步的任务，主线程不阻塞。
     * 1. runAsync 异步没有返回值
     * 2. supplyAsync 有返回值
     */
    @Test
    public void test02() {
        // 异步执行： CompletableFuture<Void> java.util.concurrent.CompletableFuture.runAsync(Runnable runnable)
        // 异步执行通过ForkJoinPool实现， 它使用守护线程去执行任务。
        // 注意这是CompletableFuture的特性， 其它CompletionStage可以override这个默认的行为
        final CompletableFuture<Void> runAsync = CompletableFuture.runAsync(() -> ThreadUtil.sleep(1));

        log.info("runAsync={}", runAsync.isDone());
        ThreadUtil.sleep(2);
        log.info("runAsync={}", runAsync.isDone());

        final CompletableFuture<String> runAsyn2 = CompletableFuture.supplyAsync(() -> Fun.applyDelay("任务1完成"));
        log.info("supplyAsync={}", runAsyn2.getNow("没有数据"));
        ThreadUtil.sleep(2);
        log.info("supplyAsync={}", runAsyn2.getNow("没有数据"));
    }

    /**
     * 增加行为：thenApply。就是当前面的任务完成之后在执行。
     * 作用：将多个任务依次同步执行。多个任务在同一个线程内依次执行
     */
    @Test
    public void test03() {
        final CompletableFuture<String> future = CompletableFuture
                .supplyAsync(() -> Fun.applyDelay("任务1完成"))
                .thenApply(result1 -> Fun.applyDelay(result1 + ", 任务2完成"))
                .thenApply(result2 -> Fun.applyDelay(result2 + ", 任务3完成"));

        log.info("开始阻塞={}", future.getNow("没有数据"));
        ThreadUtil.sleep(4);
        log.info(future.getNow("没有数据"));

    }

    /**
     * 增加行为：thenApplyAsync。就是当前面的任务完成之后在执行。
     * 与 thenApply 作用一样，但是 thenApplyAsync 是提交给线程池执行。多个任务可能在不同的线程内执行
     */
    @Test
    public void test04() {

        final CompletableFuture<String> future = CompletableFuture
                .completedFuture("main")
                .thenApplyAsync(res -> Fun.applyDelay(res + ",任务1"))
                .thenApplyAsync(res -> Fun.applyDelay(res + ",任务2"))
                .thenApplyAsync(res -> Fun.applyDelay(res + ",任务3"));
        log.info("开始阻塞={}", future.getNow("没有数据"));
        ThreadUtil.sleep(4);
        log.info(future.getNow("没有数据"));
    }

    /**
     * 提供一个Executor来异步地执行CompletableFuture
     */
    @Test
    public void test05() {
        final CompletableFuture<String> future = CompletableFuture.completedFuture("abc")
                .thenApplyAsync(msg -> {
                    ThreadUtil.sleep(4);
                    return msg.toUpperCase();
                }, executor);

        log.info("开始阻塞={}", future.getNow("没有数据"));
        ThreadUtil.sleep(5);
        log.info(future.getNow("没有数据"));
    }

    /**
     * thenAccept 接收上一个任务的执行结果作为入参，但是没有返回值
     * thenRun    无入参，也没有返回值
     */
    @Test
    public void test06() throws ExecutionException, InterruptedException {
        // 如果下一阶段接收了当前阶段的结果，但是在计算的时候不需要返回值(它的返回类型是void)， 那么它可以不应用一个函数，而是一个消费者， 调用方法也变成了thenAccept
        final StringBuilder result = new StringBuilder();
        CompletableFuture<Void> future = CompletableFuture
                .supplyAsync(() -> Fun.applyDelay("任务1完成"))
                .thenAccept(s -> {
                    log.info("thenAccept");
                    ThreadUtil.sleep(1);
                    result.append(s);
                })
                .thenRun(() -> {
                    log.info("thenRun");
                    ThreadUtil.sleep(1);
                    result.append(", thenRun");
                });
        // 使用get阻塞任务
        future.get();
        ThreadUtil.sleep(3);
        log.info(result.toString());
    }

    @Test
    public void test07() {
        // 在test06 上，使用异步
        final StringBuilder result = new StringBuilder();
        CompletableFuture.completedFuture("abc").thenAcceptAsync(s -> {
            ThreadUtil.sleep(2);
            result.append(s);
            log.info("end");
        });
        log.info(result.toString());
        ThreadUtil.sleep(3);
        log.info(result.toString());
    }

    /**
     * jdk9特性。延迟执行：CompletableFuture.delayedExecutor
     */
    @Test
    public void test08() {
        // 异步执行，使用CompletableFuture.delayedExecutor延迟一定时间后，执行操作
//        final CompletableFuture<String> future = CompletableFuture.completedFuture("abc")
//                .thenApplyAsync(String::toUpperCase, CompletableFuture.delayedExecutor(2, TimeUnit.SECONDS));
//        log.info(future.getNow(null));
//        ThreadUtil.sleep(3);
//        log.info(future.getNow(null));
    }

    @Test
    public void test09() {
//        // 执行操作的时候，出现异常
//        final CompletableFuture<String> future = CompletableFuture.completedFuture("abc")
//                .thenApplyAsync(Fun::applyExp, CompletableFuture.delayedExecutor(2, TimeUnit.SECONDS));
//        // 执行操作的时候，正常完成
//        // CompletableFuture<String> future = CompletableFuture.completedFuture("abc").thenApplyAsync(Fun::apply, CompletableFuture.delayedExecutor(2, TimeUnit.SECONDS));
//
//        // 显示的获得任务的执行状态，msg 是任务返回的数据，throwable是任务执行中不抓到的异常
//        final CompletableFuture<String> exceptionHandler = future
//                .handle((msg, throwable) -> (throwable != null) ? "test09 异常了，退出任务" : "任务正常");
//        try {
//            log.info(future.join());
//        } catch (final RuntimeException e) {
//            e.printStackTrace();
//        }
//        log.info(exceptionHandler.join());
    }

    // 取消任务
    @Test
    public void test10() {
//        final CompletableFuture<String> future = CompletableFuture.completedFuture("abc")
//                .thenApplyAsync(Fun::apply, CompletableFuture.delayedExecutor(2, TimeUnit.SECONDS));
//        // 显示的处理异常
//        final CompletableFuture<String> exceptionHandler = future
//                .handle((msg, throwable) -> (throwable != null) ? "test10 任务取消了，退出任务" : "任务正常");
//
//        // 方式1
//        try {
//            future.cancel(true);// 取消任务
//        } catch (final Exception e) {
//            e.printStackTrace();
//        }
        // 方式二
//		try {
//			future.completeExceptionally(new RuntimeException("手动抛出异常，让任务终止"));
//		} catch (Exception e) {
//			e.printStackTrace();
//		}
//        log.info(exceptionHandler.join());
    }

    @Test
    public void test11() {
        log.info("begin");
        // future1
        final CompletableFuture<String> future1 = CompletableFuture.completedFuture("abc").thenApplyAsync(msg -> {
            log.info("future1开始");
            ThreadUtil.sleep(5);
            log.info("future1完成");
            return msg.toUpperCase();
        });
        // future2
        final CompletableFuture<String> future2 = CompletableFuture.completedFuture("def").thenApplyAsync(msg -> {
            log.info("future2开始");
            ThreadUtil.sleep(1);
            log.info("future2完成");
            return msg.toUpperCase();
        });
        // future3
        // future1 和 future2 中有任意一个完成后，开始执行future3，调用功能函数
        final CompletableFuture<String> future3 = future1.applyToEither(future2, msg1 -> {
            log.info("future3");
            return msg1.toUpperCase();
        });

        log.info(future3.join());
    }

    // 与test11相反,任意任务完成后，调用消费函数
    @Test
    public void test12() {
        log.info("begin");
        // future1
        final CompletableFuture<String> future1 = CompletableFuture.completedFuture("abc").thenApplyAsync(msg -> {
            log.info("future1开始");
            ThreadUtil.sleep(2);
            log.info("future1完成");
            return msg.toUpperCase();
        });
        // future2
        final CompletableFuture<String> future2 = CompletableFuture.completedFuture("def").thenApplyAsync(msg -> {
            log.info("future2开始");
            ThreadUtil.sleep(1);
            log.info("future2完成");
            return msg.toUpperCase();
        });
        // future3
        // future1 和 future2 中有任意一个完成后，开始执行future3
        final StringBuilder result = new StringBuilder();
        future1.acceptEither(future2, result::append);

        ThreadUtil.sleep(3);
        log.info(result.toString());
    }

    // 两个任务都完成后，执行新的线程
    @Test
    public void test13() {
        log.info("begin");
        // future1
        final CompletableFuture<String> future1 = CompletableFuture.completedFuture("abc").thenApplyAsync(msg -> {
            log.info("future1开始");
            ThreadUtil.sleep(2);
            log.info("future1完成");
            return msg.toUpperCase();
        });
        // future2
        final CompletableFuture<String> future2 = CompletableFuture.completedFuture("def").thenApplyAsync(msg -> {
            log.info("future2开始");
            ThreadUtil.sleep(1);
            log.info("future2完成");
            return msg.toUpperCase();
        });
        // 执行新的线程
        final CompletableFuture<Void> future3 = future1.runAfterBoth(future2, () -> {
            log.info("future3");
        });
        System.out.println(future3.join());
    }

    // 两个任务都完成后，执行BiConsumer函数，处理 future1和future2的结果
    @Test
    public void test14() {
        log.info("begin");
        // future1
        final CompletableFuture<String> future1 = CompletableFuture.completedFuture("abc").thenApplyAsync(msg -> {
            log.info("future1开始");
            ThreadUtil.sleep(2);
            log.info("future1完成");
            return msg.toUpperCase();
        });
        // future2
        final CompletableFuture<String> future2 = CompletableFuture.completedFuture("def").thenApplyAsync(msg -> {
            log.info("future2开始");
            ThreadUtil.sleep(1);
            log.info("future2完成");
            return msg.toUpperCase();
        });

        final StringBuilder result = new StringBuilder();
        // thenAcceptBoth 			同步处理
        // thenAcceptBothAsync 		异步处理
        final CompletableFuture<Void> future3 = future1
                .thenAcceptBoth(future2, (f1, f2) -> result.append(f1).append("~").append(f2));
        future3.join();

        log.info(result.toString());
    }

    // 两个任务都完成后，执行BiFunction函数，处理 future1和future2的结果
    @Test
    public void test15() {
        log.info("begin");
        // future1
        final CompletableFuture<String> future1 = CompletableFuture.completedFuture("abc").thenApplyAsync(msg -> {
            log.info("future1开始");
            ThreadUtil.sleep(2);
            log.info("future1完成");
            return msg.toUpperCase();
        });
        // future2
        final CompletableFuture<String> future2 = CompletableFuture.completedFuture("def").thenApplyAsync(msg -> {
            log.info("future2开始");
            ThreadUtil.sleep(1);
            log.info("future2完成");
            return msg.toUpperCase();
        });
        // thenCombine 			同步处理
        // thenCombineAsync 	异步处理
        final CompletableFuture<String> future3 = future1.thenCombine(future2, (f1, f2) -> f1 + "~" + f2);
        log.info(future3.join());
    }

    // 多个任务执行，只要有一个任务完成就执行下面的操作
    @Test
    public void test16() {
        log.info("begin");
        final List<String> messages = Lists.newArrayList("future1", "future2", "future3");
        // 创建多个任务
        final List<CompletableFuture<String>> futureList = messages.stream()
                .map(msg -> CompletableFuture.completedFuture(msg).thenApplyAsync(Fun::applyRandom))
                .collect(Collectors.toList());

        final StringBuilder result = new StringBuilder();
        final CompletableFuture<Object> future = CompletableFuture
                // 任意一个完成时，执行下面的操作
                .anyOf(futureList.toArray(new CompletableFuture[futureList.size()]))
                .whenComplete((msg, throwable) -> {
                    if (throwable == null) {
                        result.append("任务完成").append(msg);
                    } else {
                        result.append("任务出现异常").append(throwable.getMessage());
                    }
                });

        future.join();
        log.info(result.toString());
    }

    // 多个任务执行，当所有任务都完成后，执行下面的操作
    @Test
    public void test17() {
        log.info("begin");
        final List<String> messages = Lists.newArrayList("future1", "future2", "future3");
        // 创建多个任务
        final List<CompletableFuture<String>> futureList = messages.stream()
                .map(msg -> CompletableFuture.completedFuture(msg).thenApplyAsync(Fun::applyRandom))
                .collect(Collectors.toList());

        final StringBuilder result = new StringBuilder();
        final CompletableFuture<Void> future = CompletableFuture
                // 所有任务都完成后
                .allOf(futureList.toArray(new CompletableFuture[futureList.size()]))
                // 同步执行下面操作
                .whenComplete((msg, throwable) -> {
                    // msg 是 null
                    futureList.forEach(cf -> result.append(cf.getNow(null)).append(","));
                });

        future.join();
        log.info(result.toString());
    }

    public static class Fun {

        public static String apply(final String s) {
            return s.toUpperCase();
        }

        public static String applyExp(final String s) {
            throw new RuntimeException("出现异常");
        }

        public static String applyDelay(final String s) {
            ThreadUtil.sleep(1);
            log.info(s);
            return s.toUpperCase();
        }

        public static String applyRandom(final String s) {
            final Random random = new Random();
            ThreadUtil.sleep(random.nextInt(10));
            return s.toUpperCase();
        }
    }

    @Data
    @Builder
    public static class Machine {

        private String instanceId;
        private String ip;
        private String status;
    }

}
